package cc.verywell.pureblock.net;

import java.io.IOException;
import java.net.SocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.util.EncryptUtil;

/**
* 
* Copyright © 2020 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2020年2月17日 上午11:27:04
*/
public class AioClientInner implements  CompletionHandler<Void,AioClientInner.ClientFuture<?>>,AioClientInterface<NormalSession,AioClientInner.ClientConfig>{

	static private NullPointerException nullNormalChannelExcetion = new NullPointerException("NormalChannel return null"); 
	private BlockingQueue<HandshakeChannelClient> handshakeChannels;

	final private AioProvider provider;		    ///连接主部件
	AsynchronousChannelGroup channelThread;     ///连接用线程
	public AioClientInner(AioProvider provider,AsynchronousChannelGroup channelThread) {
		Objects.requireNonNull(provider);
		this.provider = provider;
		this.channelThread = channelThread;
		this.handshakeChannels = new LinkedBlockingQueue<>(AioProvider.MAX_HANDSHAKE_CHANNEL_CACHE);
	}
	
	@FunctionalInterface
	static public interface ClientConfig{
		public SocketAddress getRemoteAddress();
		default public SocketAddress getLocalBindAddress() {return null;}
		default public void beforeConnect(AsynchronousSocketChannel channel) {}
	};
	
	/////////////第一站
	public <A> Future<NormalSession> connect(ClientConfig remote,A attachment,CompletionHandler<NormalSession, A> handler){
		Objects.requireNonNull(remote);
		///以上会自己throwException或者返回null,往下出错只会执行failed;
		AsynchronousSocketChannel ch = null;
		Throwable che = null;
		try {
			AsynchronousChannelGroup acg = this.channelThread;
			if(acg==null) {
				synchronized (this) {
					acg = this.channelThread;
					if(acg==null) {
						acg = this.channelThread = provider.getDefaultChannelGroup(0);  ///用默认线程连接;
						if(acg==null)throw new NullPointerException("cannot get default provider thread.");
					}
				}
			}
			ch = AsynchronousSocketChannel.open(acg);
		}catch(Throwable e) {
			che = e;
		}
		if(ch==null) {
			if(handler!=null)handler.failed(che, attachment);
			return canceledFuture;
		}
		//获得了channel，预备设置,然后连接
		ClientFuture<A> f = new ClientFuture<A>(ch, attachment, handler);
		try {
			provider.pendingSet.add(ch);
			SocketAddress bindAddress = remote.getLocalBindAddress();
			if(bindAddress!=null) {
				ch.bind(bindAddress);
			}
			ch.setOption(StandardSocketOptions.SO_REUSEADDR, true);
			ch.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
			remote.beforeConnect(ch); //在这里标记开始时间;
			ch.connect(remote.getRemoteAddress(),f,this);///进入第二站
			ch = null;//仅仅是作个成功标记
		} catch (Throwable e) {
			che = e;
		} finally {
			if(ch!=null) {
				f.cancel(che);
			}
		}
		return f;
	}
	
	
	/////////第二站
	@Override
	public void completed(Void __, AioClientInner.ClientFuture<?> f) {
		HandshakeChannelClient hc;
		try {
			hc = handshakeChannels.poll();
			if(hc==null)hc = new HandshakeChannelClient(provider.myAlias()); //这里会throw NoSuchAlgorithmException
			hc.init(f.channel, AioProvider.TIME_OUT, f); //进入第三站
		} catch (Throwable e) {
			////hc如果出了问题就扔掉不用了;
			//handshakeChannels.offer(hc.clearUser());
			f.cancel(e);
		}
	}
	@Override
	public void failed(Throwable exc, AioClientInner.ClientFuture<?> f) {
		f.cancel(exc);
	}
	
	
	public static Future<NormalSession> canceledFuture = new Future<NormalSession>() {
		@Override
		public NormalSession get() throws InterruptedException, ExecutionException {return null;}
		@Override
		public NormalSession get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {return null;}
		@Override
		public boolean cancel(boolean mayInterruptIfRunning) {return false;}
		@Override
		public boolean isCancelled() {return true;	}
		@Override
		public boolean isDone() {return true;}
	};
	class ClientFuture<A> implements Future<NormalSession>,HandshakeFunction<HandshakeChannelClient>{
		private NormalSession normalSession;
		private boolean isCancelled,isDone;
		
		//以下内容会逐步清空
		private AsynchronousSocketChannel channel; ///会在第三站清除
		private NormalChannel normalcl; ///会在第三站建立，在完成时清空
		private A attachment; //最后清除
		private CompletionHandler<NormalSession, A> handler; //最后清除
		
		ClientFuture(AsynchronousSocketChannel ch,A attachment,CompletionHandler<NormalSession, A> handler){
			this.channel = ch;
			this.attachment = attachment;
			this.handler = handler;
		}
		
		////Future功能区 - Future<NormalSession>
		@Override
		public boolean cancel(boolean mayInterruptIfRunning) {
			return cancel(null);
		}
		@Override
		public boolean isCancelled() {
			return isCancelled;
		}

		@Override
		public boolean isDone() {
			return isDone;
		}

		@Override
		synchronized public NormalSession get() throws InterruptedException, ExecutionException {
			if(isDone)return normalSession;
			wait();
			return normalSession;
		}

		@Override
		synchronized public NormalSession get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
			if(isDone)return normalSession;
			unit.timedWait(this, timeout);
			if(isDone)return normalSession;
			throw new TimeoutException();
		}
		
		/////////第三站 - HandshakeFunction<HandshakeChannelClient>
		@Override
		public void handshakeCompleted(HandshakeChannelClient obj, EncryptUtil eu, ByteBuffer unread, int salt) {
			try {
				NormalChannel nc = provider.handshakeCompleted(obj.getChannel(), eu, unread, new ClientNormalFunction<A>(this), salt, true);//进入第四站
				if(nc==null) { ////致命错误;
					cancel(nullNormalChannelExcetion);
				}else {
					synchronized (this) {
						channel = null;
						///于此同时pendingSet也处理掉了;
						normalcl = nc; //保留这个是为了能随时使用cancel选项终止进程;
					}
				}
			}finally {
				handshakeChannels.offer(obj.clearUser());
			}
		}
		@Override
		public void handshakeFailed(HandshakeChannelClient obj, Throwable exp) {
			try {
				provider.handshakeFaild(obj.getChannel(), exp);
				cancel(exp);
			}finally {
				handshakeChannels.offer(obj.clearUser());
			}
		}

		////第四站在最后面		
		
		/////////第五站 - 完成
		public void done(NormalSession session) {
			A attachment;
			CompletionHandler<NormalSession, A> handler;
			synchronized (this) {
				if(isDone)return;
				isDone = true;
				this.normalSession = session;
				this.normalcl = null;
				handler = this.handler;
				attachment = this.attachment;
				this.handler = null;
				this.attachment = null;
				this.notifyAll();
			}
			if(handler!=null) {
				handler.completed(session, attachment);
			}
		}
		public boolean cancel(Throwable e) {
			A attachment;
			CompletionHandler<NormalSession, A> handler;
			synchronized (this) {
				if(isDone)return false;
				isCancelled = isDone = true;
				if(channel!=null) {
					if(channel.isOpen()) {
						try {
							channel.shutdownInput();
						} catch (IOException ioe) {
							if(Pblog.IO)Pblog.IO(this,"@closing: ",ioe);
						}
						try {
							channel.shutdownOutput();
						} catch (IOException ioe) {
							if(Pblog.IO)Pblog.IO(this,"@closing: ",ioe);
						}
					}
					try {
						channel.close();
					} catch (IOException ioe) {
						if(Pblog.IO)Pblog.IO(this,"@closing: ",ioe);
					}
					provider.pendingSet.remove(channel);
					channel = null;
				}
				if(normalcl!=null) {
					normalcl.close();
					normalcl=null;
				}
				handler = this.handler;
				attachment = this.attachment;
				this.handler = null;
				this.attachment = null;
				this.notifyAll();
			}
			if(handler!=null) {
				handler.failed(e, attachment);
			}
			return true;
		}
		
		@Override
		public String toString() {
			StringBuffer sb = new StringBuffer();
			if(normalSession!=null) {
				return "connected "+normalSession.toString();
			}
			sb.append("connecting").append("[").append(provider.myAlias()).append("->");
			AsynchronousSocketChannel channel = this.channel;
			if(channel!=null) {
				sb.append("(").append(Integer.toHexString(channel.hashCode())).append(")");
				try {
					sb.append(channel.getRemoteAddress() );
				} catch (IOException e) {
					sb.append("ErrAddress").append("<").append(e.getLocalizedMessage()).append(">");
				}
			}else{
				if(isCancelled)sb.append("cancelled");
				else if(isDone)sb.append("done");
				else if(normalcl!=null) {
					try {
						sb.append(normalcl.getChannel().getRemoteAddress());
					} catch (IOException e) {
						sb.append("ErrNormalChannel").append("<").append(e.getLocalizedMessage()).append("> in step 4");
					}
				}else sb.append("(lost channel)");
			}
			sb.append("]").append(Integer.toHexString(hashCode()));
			return sb.toString();
		}
		
	}
	
	/////////第四站 - NormalFunction
	static class ClientNormalFunction<A> implements NormalFunction{ 
		private AioClientInner.ClientFuture<A> clientFuture;
		private NormalFunction nextfunction;
		public ClientNormalFunction(AioClientInner.ClientFuture<A> f) {
			this.clientFuture = f;
		}
		@Override
		public void addNext(NormalFunction function) {
			this.nextfunction = function;
		}
		@Override
		public void init(NormalSession channel) {
			if(nextfunction!=null)nextfunction.init(channel);
			AioClientInner.ClientFuture<A> f = clientFuture;
			if(f!=null) {
				clientFuture = null;
				f.done(channel);//进入第五站
			}
		}
		@Override
		public void closed(NormalChannel channel) {
			if(nextfunction!=null)nextfunction.closed(channel);
			AioClientInner.ClientFuture<A> f = clientFuture;
			if(f!=null) {
				clientFuture = null;
				f.cancel(null);//进入第五站
			}
		}

	}



}
